<?php
/*
 * @ https://EasyToYou.eu - IonCube v11 Decoder Online
 * @ PHP 7.2 & 7.3
 * @ Decoder version: 1.1.6
 * @ Release: 10/08/2022
 */

// Decoded file for php version 71.
namespace app\Common;

class PublisherApi
{
    protected $configs;
    protected $connection;
    protected $exchangeName;
    protected $queueName;
    protected $routingKey;
    protected $qos_limit = 20;
    public function __construct()
    {
        $this->configs = ["host" => \think\facade\Config::get("rabbit.rabbit_host"), "port" => \think\facade\Config::get("rabbit.rabbit_port"), "login" => \think\facade\Config::get("rabbit.rabbit_login"), "password" => \think\facade\Config::get("rabbit.rabbit_passwd"), "vhost" => \think\facade\Config::get("rabbit.rabbit_vhost")];
        $this->qos_limit = \think\facade\Config::get("rabbit.rabbit_qos_limit");
        $this->exchangeName = \think\facade\Config::get("rabbit.rabbit_exchange_name");
        $this->queueName = \think\facade\Config::get("rabbit.rabbit_query_name");
        $this->routingKey = \think\facade\Config::get("rabbit.rabbit_key");
    }
    public function publishMessage($message)
    {
        $this->connection = new \PhpAmqpLib\Connection\AMQPStreamConnection($this->configs["host"], $this->configs["port"], $this->configs["login"], $this->configs["password"], $this->configs["vhost"]);
        $connection = $this->connection;
        $channel = $connection->channel();
        $channel->exchange_declare($this->exchangeName, "direct", false, false, false);
        $channel->queue_declare($this->queueName, false, true, false, false, false);
        $channel->queue_bind($this->queueName, $this->exchangeName, $this->routingKey);
        $msg = new \PhpAmqpLib\Message\AMQPMessage($message, ["delivery_mode" => \PhpAmqpLib\Message\AMQPMessage::DELIVERY_MODE_PERSISTENT]);
        $channel->basic_publish($msg, $this->exchangeName, $this->routingKey);
        $channel->close();
        $connection->close();
    }
    public function delayMessage($message, $expiration = 10000)
    {
        $this->connection = new \PhpAmqpLib\Connection\AMQPStreamConnection($this->configs["host"], $this->configs["port"], $this->configs["login"], $this->configs["password"], $this->configs["vhost"]);
        $connection = $this->connection;
        $channel = $connection->channel();
        $cache_exchange_name = "cache_exchange" . $expiration;
        $cache_queue_name = "cache_queue" . $expiration;
        $channel->exchange_declare($this->exchangeName, "direct", false, false, false);
        $channel->exchange_declare($cache_exchange_name, "direct", false, false, false);
        $tale = new \PhpAmqpLib\Wire\AMQPTable();
        $tale->set("x-dead-letter-exchange", $this->exchangeName);
        $tale->set("x-dead-letter-routing-key", $this->routingKey);
        $tale->set("x-message-ttl", $expiration);
        $channel->queue_declare($cache_queue_name, false, true, false, false, false, $tale);
        $channel->queue_bind($cache_queue_name, $cache_exchange_name, "");
        $channel->queue_declare($this->queueName, false, true, false, false, false);
        $channel->queue_bind($this->queueName, $this->exchangeName, $this->routingKey);
        $msg = new \PhpAmqpLib\Message\AMQPMessage($message, ["delivery_mode" => \PhpAmqpLib\Message\AMQPMessage::DELIVERY_MODE_PERSISTENT]);
        $channel->basic_publish($msg, $cache_exchange_name, "");
        $channel->close();
        $connection->close();
    }
    public function scheduleMessage($message, $taskName, $expiration = 10000)
    {
        $this->connection = new \PhpAmqpLib\Connection\AMQPStreamConnection($this->configs["host"], $this->configs["port"], $this->configs["login"], $this->configs["password"], $this->configs["vhost"]);
        $connection = $this->connection;
        $channel = $connection->channel();
        $schedule_exchange_name = "schedule_exchange_" . $taskName;
        $schedule_queue_name = "schedule_queue_" . $taskName;
        $channel->exchange_declare($this->exchangeName, "direct", false, false, false);
        $channel->exchange_declare($schedule_exchange_name, "direct", false, false, false);
        $tale = new \PhpAmqpLib\Wire\AMQPTable();
        $tale->set("x-dead-letter-exchange", $this->exchangeName);
        $tale->set("x-dead-letter-routing-key", $this->routingKey);
        $tale->set("x-message-ttl", $expiration);
        $channel->queue_declare($schedule_queue_name, false, false, false, false, false, $tale);
        $channel->queue_bind($schedule_queue_name, $schedule_exchange_name, "");
        $channel->queue_declare($this->queueName, false, true, false, false, false);
        $channel->queue_bind($this->queueName, $this->exchangeName, $this->routingKey);
        $msg = new \PhpAmqpLib\Message\AMQPMessage($message, ["delivery_mode" => \PhpAmqpLib\Message\AMQPMessage::DELIVERY_MODE_PERSISTENT]);
        $channel->basic_publish($msg, $schedule_exchange_name, "");
        $channel->close();
        $connection->close();
    }
    public function scheduleInfo($taskName, $expiration = 10000)
    {
        $this->connection = new \PhpAmqpLib\Connection\AMQPStreamConnection($this->configs["host"], $this->configs["port"], $this->configs["login"], $this->configs["password"], $this->configs["vhost"]);
        $connection = $this->connection;
        $channel = $connection->channel();
        $schedule_queue_name = "schedule_queue_" . $taskName;
        $tale = new \PhpAmqpLib\Wire\AMQPTable();
        $tale->set("x-dead-letter-exchange", $this->exchangeName);
        $tale->set("x-dead-letter-routing-key", $this->routingKey);
        $tale->set("x-message-ttl", $expiration);
        $result = $channel->queue_declare($schedule_queue_name, false, false, false, false, false, $tale);
        $channel->close();
        $connection->close();
        $result = $result ?: [0, 0];
        return ["ready" => $result[1], "unacked" => $result[2]];
    }
    public function consumer()
    {
        $this->connection = new \PhpAmqpLib\Connection\AMQPStreamConnection($this->configs["host"], $this->configs["port"], $this->configs["login"], $this->configs["password"], $this->configs["vhost"], false, "AMQPLAIN", NULL, "en_US", 3, 360, NULL, false, 180);
        $connection = $this->connection;
        $channel = $connection->channel();
        $channel->exchange_declare($this->exchangeName, "direct", false, false, false);
        $channel->queue_declare($this->queueName, false, true, false, false, false);
        $channel->queue_bind($this->queueName, $this->exchangeName, $this->routingKey);
        $callback = function ($msg) {
            messagesProcess($msg);
        };
        $channel->basic_qos(NULL, $this->qos_limit, NULL);
        $channel->basic_consume($this->queueName, "", false, false, false, false, $callback);
        while (count($channel->callbacks)) {
            $channel->wait();
        }
        $channel->close();
        $connection->close();
    }
}

?>